package bsp.scan;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hama.HamaConfiguration;
import org.apache.hama.bsp.BSPJob;
import org.apache.hama.bsp.FileInputFormat;
import org.apache.hama.bsp.FileOutputFormat;
import org.apache.hama.bsp.TextInputFormat;
import org.apache.hama.bsp.TextOutputFormat;

import bsp.scan.TestBSPScan.MyScan;

import util.functors.SumLong;

public class Scan extends BspScanHelper {
	public static final Log LOG = LogFactory.getLog(Scan.class);

	public void setOperator() {
		this.oplus = new SumLong();
	}
	
	public static BSPJob createScanJob( final Path in, final Path out) throws IOException{
		HamaConfiguration conf = new HamaConfiguration();

		BSPJob job = new BSPJob(conf);
		////
		job.setInputFormat(TextInputFormat.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);
		job.setOutputFormat(TextOutputFormat.class);
		FileInputFormat.setInputPaths(job, in);
		FileOutputFormat.setOutputPath(job, out);
		
		return job;
	}
}